package com.cloudera;

import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;

import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;

/**
 * Created by shuyun on 2016/8/25.
 */
public class DateCompare extends UDF {
    private static final DateFormat DATE_FORMAT = new SimpleDateFormat("yyyy-MM-dd");

    /**
     * 时间比较
     * @param interval          比较类型（年、月、日）
     * @param datetime_expr1   比较日期
     * @param datetime_expr2   比较日期
     * @param plan               计划
     * @return
     * @throws UDFArgumentException
     */
    public LongWritable evaluate(Text interval, Text datetime_expr1, Text datetime_expr2, IntWritable plan) throws UDFArgumentException {

        if(null == datetime_expr1 || null == datetime_expr2) return null;

        DateFix dateFix;
        Calendar cal1 = Calendar.getInstance();
        Calendar cal2 = Calendar.getInstance();

        try {
            dateFix = DateFix.valueOf(interval.toString().toUpperCase());

            String dateCompare = null;
            String dateCompareTo = null;
            if(plan.get()==1) {
                dateCompare = datetime_expr1.toString();
                dateCompareTo = datetime_expr2.toString();
            } else if(plan.get()==2) {
                dateCompare = datetime_expr2.toString();
                dateCompareTo = datetime_expr1.toString();
            }
            cal1.setTime(DATE_FORMAT.parse(dateCompare));
            cal2.setTime(DATE_FORMAT.parse(dateCompareTo));
        } catch (IllegalArgumentException iae) {
            throw new UDFArgumentException("NotSupport");
        } catch (ParseException pe) {
            throw new UDFArgumentException("Input Argument Of Date Must Be \"yyyy-MM-dd\"");
        }

        int day = cal2.get(Calendar.DAY_OF_MONTH) - cal1.get(Calendar.DAY_OF_MONTH);
        int month = cal2.get(Calendar.MONTH) - cal1.get(Calendar.MONTH);
        int year = cal2.get(Calendar.YEAR) - cal1.get(Calendar.YEAR);

        switch (dateFix) {
            case DAY: {
                return new LongWritable((cal2.getTimeInMillis() - cal1.getTimeInMillis()) / (24 * 60 * 60 * 1000));
            }
            case MONTH: {
                if(year > 0 && day < 0) {
                    month = (cal2.get(Calendar.MONTH) + 12 - cal1.get(Calendar.MONTH)) - 1;
                    year--;
                } else if(year < 0 && day > 0) {
                    month = (-1 * cal1.get(Calendar.MONTH) - 12 + cal2.get(Calendar.MONTH)) + 1;
                    year++;
                } else if (year == 0) {
                    if(month > 0 && day < 0) {
                        month--;
                    } else if(month < 0 && day > 0) {
                        month++;
                    }
                }

                return  new LongWritable((year * 12 + month));
            }
            default: {
                if(year > 0 && (month < 0 || (month == 0 && day < 0))) year--;
                return  new LongWritable(year);
            }
        }
    }
}
